/*
 * Copyright 2012 Gregor Schauer
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package at.schauer.gregor.vienna.datastore.remote;

import at.schauer.gregor.vienna.context.ViennaContext;
import at.schauer.gregor.vienna.context.ViennaContextAware;
import at.schauer.gregor.vienna.datastore.DataStore;
import at.schauer.gregor.vienna.exception.ViennaException;
import org.apache.commons.lang3.ClassUtils;

import javax.annotation.Nullable;
import java.rmi.NotBoundException;
import java.rmi.Remote;
import java.rmi.RemoteException;
import java.rmi.registry.LocateRegistry;
import java.rmi.registry.Registry;
import java.rmi.server.UnicastRemoteObject;
import java.util.Collection;
import java.util.HashSet;
import java.util.Set;

import static at.schauer.gregor.vienna.serializer.Serializer.CLASS_OBJECT_SEPARATOR;

/**
 * @author Gregor Schauer
 */
public class RMIDataStore<T extends Remote> implements DataStore<T>, ViennaContextAware {
	protected ViennaContext ctx;
	protected Registry registry;

	public RMIDataStore() throws RemoteException {
		this(null, Registry.REGISTRY_PORT);
	}

	public RMIDataStore(@Nullable String host, int port) throws RemoteException {
		try {
			registry = LocateRegistry.getRegistry(host, port);
		} catch (RemoteException e) {
			registry = LocateRegistry.createRegistry(port);
		}
	}

	@Override
	@SuppressWarnings("unchecked")
	public <E extends T> E load(Class<E> type, Object identifier) {
		try {
			return (E) registry.lookup(getName(type, identifier));
		} catch (Exception e) {
			throw new ViennaException(e);
		}
	}

	@Override
	public Object save(T obj) {
		try {
			String name = getName(obj);
			registry.rebind(name, UnicastRemoteObject.exportObject(obj, 0));
			return name;
		} catch (RemoteException e) {
			throw new ViennaException(e);
		}
	}

	@Override
	public void delete(T obj) {
		try {
			registry.unbind(getName(obj));
		} catch (RemoteException e) {
			throw new ViennaException(e);
		} catch (NotBoundException e) {
			throw new ViennaException(e);
		}
	}

	@Override
	@SuppressWarnings("unchecked")
	public <E extends T> Collection<E> list(Class<E> type) {
		try {
			String typeName = getName(type, null);
			Set<E> localObjects = new HashSet<E>();
			String[] names = registry.list();
			for (String name : names) {
				String className = name.substring(0, name.indexOf(CLASS_OBJECT_SEPARATOR));
				Class<E> aClass = (Class<E>) ClassUtils.getClass(className);
				String id = name.substring(name.indexOf(CLASS_OBJECT_SEPARATOR) + 1);
				if (name.startsWith(typeName) || type.isAssignableFrom(aClass)) {
					Remote remote = load(aClass, id);
					if (type.isAssignableFrom(remote.getClass())) {
						localObjects.add((E) remote);
					}
				}
			}
			return localObjects;
		} catch (RemoteException e) {
			throw new ViennaException(e);
		} catch (ClassNotFoundException e) {
			throw new ViennaException(e);
		}
	}

	@SuppressWarnings("unchecked")
	protected String getName(T obj) {
		Class<T> type = (Class<T>) obj.getClass();
		return getName(type, ctx.getMetadataProvider(type).getIdentifier(obj));
	}

	protected String getName(Class<? extends T> clazz, @Nullable Object id) {
		return clazz.getName() + CLASS_OBJECT_SEPARATOR + (id != null ? id : "");
	}

	@Override
	public void setViennaContext(ViennaContext viennaContext) {
		this.ctx = viennaContext;
	}
}
